package processor

import (
	"babydoge/config/db"
	"babydoge/enum/fileds"
	"babydoge/models"
	"encoding/json"
	"gorm.io/gorm"
)

// mysql存储日志
type LogSink struct {
	client         *gorm.DB
	logChan        chan *JobLog
	autoCommitChan chan *LogBatch
}

var (
	// 单例
	G_logSink *LogSink
)

// 日志存储协程
func (logSink *LogSink) writeLoop() {
	var (
		log        *JobLog
		jobLog     *models.Queue
		jobLogByte []byte
		status     int
		err        error
	)

	for {
		select {
		case log = <-logSink.logChan:
			if log.Err == "" {
				status = fileds.Succeed
			} else {
				status = fileds.Fail
			}
			if jobLogByte, err = json.Marshal(log); err != nil {
				return
			}

			jobLog = &models.Queue{
				Log:    string(jobLogByte),
				Status: status,
			}
			logSink.client.Model(&models.Queue{
				ID: log.JobName,
			}).UpdateColumns(jobLog)
		}
	}
}

func InitLogSink() (err error) {
	var (
		//client *gorm.DB
		//dsn string
		dataBase *db.DB
	)
	//dsn = config.G_config.DbUser + ":" + config.G_config.DbPassword + "@tcp(" + config.G_config.DbHost + ":" + config.G_config.DbPort + ")/" + config.G_config.DbName + "?charset=utf8mb4&parseTime=True&loc=Local"
	//if client, err = gorm.Open(mysql.Open(dsn), &gorm.Config{
	//	Logger: logger.Default.LogMode(logger.Info),
	//}); err != nil {
	//	fmt.Println("连接db报错：", err.Error())
	//	panic(err)
	//}

	if dataBase, err = db.NewDB(); err != nil {
		panic("数据库连接失败")
	}

	//   选择db和collection
	G_logSink = &LogSink{
		client:         dataBase.Db,
		logChan:        make(chan *JobLog, 1000),
		autoCommitChan: make(chan *LogBatch, 1000),
	}

	// 启动一个mongodb处理协程
	go G_logSink.writeLoop()
	return
}

// 发送日志
func (logSink *LogSink) Append(jobLog *JobLog) {
	select {
	case logSink.logChan <- jobLog:
	default:
		// 队列满了就丢弃
	}
}
